package org.example.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @Author: tang
 * @Description:
 * @Date 2025/2/16 01:14
 */
public class FlinkSourceDataGen {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(2);
        DataGeneratorSource<String> dataSource = new DataGeneratorSource<String>(
        (
                GeneratorFunction<Long, String>) aLong -> "date: " + aLong,
                100,
                RateLimiterStrategy.perSecond(1),
                Types.STRING
        );

        environment
                .fromSource(dataSource, WatermarkStrategy.noWatermarks(),"gen-source")
                .print()
        ;

        environment.execute();
    }

}
